package com.xuhai

import java.io.{BufferedWriter, FileWriter}
import java.util.Date

import com.alibaba.fastjson.JSON
import com.xuhai.entity.{CityResult, StationResult, UnitResult}
import com.xuhai.util.{HttpClientUtils, IsPointInPolygon}
import org.apache.commons.lang.time.DateUtils
import org.apache.commons.lang3.time.FastDateFormat
import org.apache.hadoop.conf.Configuration
import org.apache.spark.sql.{SQLContext, SparkSession}

import scala.collection.JavaConversions._
import scala.collection.mutable.ListBuffer

/**
  * Created by chuxw on 2018/4/4.
  */
object ComputeEnterRate {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder.appName("enterRate").getOrCreate()
    import org.apache.phoenix.spark.SparkSqlContextFunctions
    val sparkSqlContext = new SparkSqlContextFunctions(spark.sqlContext)

    //返回结果对象
    val cityResultList = new java.util.ArrayList[CityResult]

    //获取场站信息
    val resultStr = HttpClientUtils.get("http://172.19.66.31:8080/gj/gj/exam/carEntryRateAction.do?action=getStationInfo")
    val cityJSONArray = JSON.parseObject(resultStr).getJSONArray("result")
    val citylen = cityJSONArray.size()

    //计算前一天
    val sdf = FastDateFormat.getInstance("yyyyMMdd")
    val previousdate: String = sdf.format(DateUtils.addDays(new Date(), -15))
    val endDate: String = sdf.format(DateUtils.addDays(new Date(), -14))
    val startDateStr: String = previousdate + "233000"
    val endDateStr: String = endDate + "000000"

    //遍历城市
    for (i <- 0 until citylen){
      val cityJSONObject =  cityJSONArray.getJSONObject(i)
      val city = cityJSONObject.getString("city")
      if(city != null && !city.equals("null") && city.equals("520100")){//贵阳用busId
        //使用spark sql
        val sqlContext = SQLContext.getOrCreate(spark.sparkContext)
        //phoenix连接配置
        val configuration = new Configuration()
        configuration.set("url", "jdbc:phoenix:slave1,slave2,slave3:2181/hbase")

        val stationJSONArray = cityJSONObject.getJSONArray("stations")
        val unitJSONArray = cityJSONObject.getJSONArray("units")

        //收集所有的busNo
        val busNoList = new ListBuffer[String]
        val unitLength:Int = unitJSONArray.size()
        for (i <- 0 until unitLength) {
          val unitJSONObject = unitJSONArray.getJSONObject(i)
          val busNos = unitJSONObject.getJSONArray("buses").toJavaList(classOf[String])
          if(busNos != null && busNos.size() > 0){
            busNos.foreach(s => {
              if(s.length == 4){
                busNoList.add("00"+s)
              }else if(s.length == 6){
                busNoList.add(s)
              }
            })
          }
        }

        var df = sparkSqlContext.phoenixTableAsDataFrame("GPS_DATA", Seq("CITY", "BUSID", "BUSNO", "GPSLAT", "GPSLON", "GPSDATETIME"), Some("CITY = '" + city + "' and BUSID in ('" + String.join("','",busNoList) + "') and GPSDATETIME > '"+ startDateStr +"' and GPSDATETIME < '"+ endDateStr +"'"), Some("slave1,slave2,slave3:2181"), null,configuration)
        df = df.repartition(4)
        df.cache()
        df.registerTempTable("gps_data_temp")


        //按场站统计
        val stationResultList = new ListBuffer[StationResult]
        val stationlen = stationJSONArray.size()
        for(i <- 0 until stationlen){
          val stationJSONObject = stationJSONArray.getJSONObject(i)
          var sum = 0
          if(stationJSONObject.getJSONArray("rings") != null && stationJSONObject.getJSONArray("rings").size() > 0){
            val stationPloygon:Array[Array[Double]] = IsPointInPolygon.transferToDouble2Array(stationJSONObject.getJSONArray("rings").getJSONArray(0).toJSONString)
            for(busNo <- busNoList){
              val busPointsDF = sqlContext.sql("select GPSLAT,GPSLON from gps_data_temp where busId = '" + busNo + "'").cache()
              val refCount = busPointsDF.count()
              if(refCount > 0){
                val count = busPointsDF.filter(row => {
                  try {
                    !IsPointInPolygon.isInPolygon(stationPloygon, BigDecimal(row(1).toString).doubleValue(), BigDecimal(row(0).toString).doubleValue())
                  } catch {
                    case ex: Exception => true
                  }
                }).count()
                if (count <= 10) {
                  sum += 1
                }
              }
            }
          }
          //计算场站结果
          val stationResult = new StationResult
          stationResult.setStationId(stationJSONObject.getString("stationId"))
          stationResult.setObjectId(stationJSONObject.getString("objectid"))
          stationResult.setNum(sum)
          stationResultList.add(stationResult)
        }



        //按企业单位统计
        val unitResultList = new ListBuffer[UnitResult]
        //收集所有的场站
        val stationPloygonList = new ListBuffer[String]
        for (i <- 0 until stationlen) {
          val stationJSONObject = stationJSONArray.getJSONObject(i)
          if(stationJSONObject.getJSONArray("rings") != null && stationJSONObject.getJSONArray("rings").size() > 0){
            stationPloygonList.add(stationJSONObject.getJSONArray("rings").getJSONArray(0).toString)
          }
        }
        for(i <- 0 until unitLength){
          val unitJSONObject = unitJSONArray.getJSONObject(i)
          val busIds = unitJSONObject.getJSONArray("buses").toJavaList(classOf[String])
          var sum = 0;
          for(busId <- busIds){
            val busPointsDF = sqlContext.sql("select GPSLAT,GPSLON from gps_data_temp where busId = '" + busId + "'")
            val refCount = busPointsDF.count()
            if(refCount > 0){
              import scala.util.control.Breaks._
              breakable{
                for(stationPloygon <- stationPloygonList){
                  val count = busPointsDF.filter(row => {
                    try {
                      !IsPointInPolygon.isInPolygon(IsPointInPolygon.transferToDouble2Array(stationPloygon), BigDecimal(row(1).toString).doubleValue(),BigDecimal(row(0).toString).doubleValue())
                    }catch {
                      case ex :Exception => true
                    }
                  }).count()
                  if(count <= 10) {
                    sum += 1
                    break
                  }
                }
              }
            }
          }

          val unitResult = new UnitResult
          unitResult.setUnitCode(unitJSONObject.getString("unitCode"))
          unitResult.setRate(sum.toString)
          unitResultList.add(unitResult)
        }

        val cityResult = new CityResult()
        cityResult.setCity(city)
        cityResult.setStations(stationResultList)
        cityResult.setUnits(unitResultList)
        cityResultList.add(cityResult)

      }
      else if(city != null && !"null".equals(city)){
          //使用spark sql
          val sqlContext = SQLContext.getOrCreate(spark.sparkContext)
          //phoenix连接配置
          val configuration = new Configuration()
          configuration.set("url", "jdbc:phoenix:slave1,slave2,slave3:2181/hbase")

          val stationJSONArray = cityJSONObject.getJSONArray("stations")
          val unitJSONArray = cityJSONObject.getJSONArray("units")

          //收集所有的busNo
          val busNoList = new ListBuffer[String]
          val unitLength:Int = unitJSONArray.size()
          for (i <- 0 until unitLength) {
            val unitJSONObject = unitJSONArray.getJSONObject(i)
            val busNos = unitJSONObject.getJSONArray("buses").toJavaList(classOf[String])
            if(busNos != null && busNos.size() > 0){
              busNoList.addAll(busNos)
            }
          }

          var df = sparkSqlContext.phoenixTableAsDataFrame("GPS_DATA", Seq("CITY", "BUSID", "BUSNO", "GPSLAT", "GPSLON", "GPSDATETIME"), Some("CITY = '" + city + "' and GPSDATETIME > '"+ startDateStr +"' and GPSDATETIME < '"+ endDateStr +"'"), Some("slave1,slave2,slave3:2181"), null,configuration)
          df = df.repartition(4)
          df.cache()
          df.registerTempTable("gps_data_temp")


          //按场站统计
          val stationResultList = new ListBuffer[StationResult]
          val stationlen = stationJSONArray.size()
          for(i <- 0 until stationlen){
            val stationJSONObject = stationJSONArray.getJSONObject(i)
            var sum = 0
            if(stationJSONObject.getJSONArray("rings") != null && stationJSONObject.getJSONArray("rings").size() > 0){
              val stationPloygon:Array[Array[Double]] = IsPointInPolygon.transferToDouble2Array(stationJSONObject.getJSONArray("rings").getJSONArray(0).toJSONString)
              //收集所有的busNo
              for(busNo <- busNoList){
                val busPointsDF = sqlContext.sql("select GPSLAT,GPSLON from gps_data_temp where busNo = '" + busNo + "'").cache()
                val refCount = busPointsDF.count()
                if(refCount > 0){
                  val count = busPointsDF.filter(row => {
                    try {
                      !IsPointInPolygon.isInPolygon(stationPloygon, BigDecimal(row(1).toString).doubleValue(), BigDecimal(row(0).toString).doubleValue())
                    } catch {
                      case ex: Exception => true
                    }
                  }).count()
                  if (count == 0) {
                    sum += 1
                  }
                }
              }
            }
            //计算场站结果
            val stationResult = new StationResult
            stationResult.setStationId(stationJSONObject.getString("stationId"))
            stationResult.setObjectId(stationJSONObject.getString("objectid"))
            stationResult.setNum(sum)
            stationResultList.add(stationResult)
          }



          //按企业单位统计
          val unitResultList = new ListBuffer[UnitResult]
          //收集所有的场站
          val stationPloygonList = new ListBuffer[String]
          for (i <- 0 until stationlen) {
            val stationJSONObject = stationJSONArray.getJSONObject(i)
            if(stationJSONObject.getJSONArray("rings") != null && stationJSONObject.getJSONArray("rings").size() > 0){
              stationPloygonList.add(stationJSONObject.getJSONArray("rings").getJSONArray(0).toString)
            }
          }
          for(i <- 0 until unitLength){
            val unitJSONObject = unitJSONArray.getJSONObject(i)
            val busNos = unitJSONObject.getJSONArray("buses").toJavaList(classOf[String])
            var sum = 0;
            for(busNo <- busNos){
              val busPointsDF = sqlContext.sql("select GPSLAT,GPSLON from gps_data_temp where busNo = '" + busNo + "'").cache()
              val refCount = busPointsDF.count()
              if(refCount > 0){
                import scala.util.control.Breaks._
                breakable{
                  for(stationPloygon <- stationPloygonList){
                    val count = busPointsDF.filter(row => {
                      try {
                        !IsPointInPolygon.isInPolygon(IsPointInPolygon.transferToDouble2Array(stationPloygon), BigDecimal(row(1).toString).doubleValue(),BigDecimal(row(0).toString).doubleValue())
                      }catch {
                        case ex :Exception => true
                      }
                    }).count()
                    if(count == 0) {
                      sum += 1
                      break
                    }
                  }
                }
              }
            }

            val unitResult = new UnitResult
            unitResult.setUnitCode(unitJSONObject.getString("unitCode"))
            unitResult.setRate(sum.toString)
            unitResultList.add(unitResult)
          }

          val cityResult = new CityResult()
          cityResult.setCity(city)
          cityResult.setStations(stationResultList)
          cityResult.setUnits(unitResultList)
          cityResultList.add(cityResult)
      }

    }
    var writer = new BufferedWriter(new FileWriter(args(0)))
    writer.append(previousdate)
    writer.append(JSON.toJSON(cityResultList).toString)
    val param: Map[String,String] = Map("action" -> "save", "data" -> JSON.toJSON(cityResultList).toString)
    val result = HttpClientUtils.post("http://172.19.66.31:8080/gj/gj/exam/carEntryRateAction.do", param)
    writer.append("============>>  " + result).append("\n")
    writer.close()
  }

}
